package org.apache.hadoop.hdfs.shortcircuit;

import java.io.FileInputStream;
import java.io.IOException;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.hdfs.ExtendedBlockId;
import org.apache.hadoop.hdfs.server.datanode.BlockMetadataHeader;
import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.nativeio.NativeIO;
import org.apache.hadoop.util.Time;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;

/**
 * ShortCircuitReplica类封装了一个短路读数据块副本的所有信息，只有获取了ShortCircuitReplica对象，才能构造BlockReaderLocal对象完成短路读操作。
 */
@InterfaceAudience.Private
public class ShortCircuitReplica {
  public static final Log LOG = LogFactory.getLog(ShortCircuitCache.class);

  /** 对应BlockId，唯一标识这个ShortCircuitReplica对象。 */
  final ExtendedBlockId key;

  /** 副本的数据块文件输入流。 */
  private final FileInputStream dataStream;

  // 不需要跟datanode再建立tcp连接读取block数据了，通过这两个本地数据流量就可以实现从本地读取block数据了

  /** 副本的校验文件输入流。 */
  private final FileInputStream metaStream;



  /** 本校验文件头 */
  private final BlockMetadataHeader metaHeader;

  /** 管理这个ShortCircuitReplica对象的ShortCircuitCache对象 */
  private final ShortCircuitCache cache;

  /**
   * Monotonic time at which the replica was created.
   */
  private final long creationTimeMs;

  /** 这个ShortCircuitReplica对象对应的共享内存中的槽位 */
  private final Slot slot;
  
  /** 当前ShortCircuitReplica对象在内存中的映射数据 */
  Object mmapData;

  /** 当前ShortCircuitReplica对象是否在cahce中被删除 */
  boolean purged = false;

  /**
   * 常重要的一个字段，标识当前ShortCircuitReplica对象被引用的次数。
   * ShortCircuitReplica只可能被ShortCircuitCache、BlockReaderLocal以及ClientMmap对象引用。
   * 当我们构建一个ShortCircuitReplica对象时，这个值为2，因为在创建时，ShortCircuitReplica对象就已经被ShortCircuitCache以及请求类引用了。
   * 只有refCount==0时，ShortCircuitReplica对象才可以被关闭。
   */
  int refCount = 2;

  /**
   *  ShortCircuitReplica对象被加入ShortCircuitCache的evictable队列的时间。
   *  如果不在evictable队列中，则为null。
   */
  private Long evictableTimeNs = null;

  public ShortCircuitReplica(ExtendedBlockId key, FileInputStream dataStream, FileInputStream metaStream,
      ShortCircuitCache cache, long creationTimeMs, Slot slot) throws IOException {
    this.key = key;
    this.dataStream = dataStream;
    this.metaStream = metaStream;
    this.metaHeader = BlockMetadataHeader.preadHeader(metaStream.getChannel());
    if (metaHeader.getVersion() != 1) {
      throw new IOException("invalid metadata header version " + metaHeader.getVersion() + ".  Can only handle version 1.");
    }
    this.cache = cache;
    this.creationTimeMs = creationTimeMs;
    this.slot = slot;
  }

  /**
   * unref()用于在不需要对ShortCircuitReplica对象的引用时，解除对ShortCircuitReplica对象的引用.
   * 这个方法调用了ShortCircuitCache的同名方法进行解关联操作
   */
  public void unref() {
    cache.unref(this);
  }

  /**
   * Check if the replica is stale.
   * Must be called with the cache lock held.
   */
  boolean isStale() {
    if (slot != null) {
      // Check staleness by looking at the shared memory area we use to communicate with the DataNode.
      boolean stale = !slot.isValid();
      return stale;
    } else {
      // Fall back to old, time-based staleness method.
      long deltaMs = Time.monotonicNow() - creationTimeMs;
      long staleThresholdMs = cache.getStaleThresholdMs();
      if (deltaMs > staleThresholdMs) {
        return true;
      } else {
        return false;
      }
    }
  }
  
  /**
   * addNoChecksumAnchor()方法用于在共享内存的槽位中添加一个免校验（no-checksum）的锚（anchor）。
   * 只有当Datanode通过mlock操作将数据块副本缓存到内存后，也就是副本对应的Slot是可锚状态时，才可以添加这个锚。
   * addNoChecksumAnchor()方法直接调用了Slot.addAnchor()方法执行这个操作。
   */
  public boolean addNoChecksumAnchor() {
    if (slot == null) {
      return false;
    }
    boolean result = slot.addAnchor();
    return result;
  }

  /**
   * Remove a no-checksum anchor for our shared memory slot.
   *
   * This method does not require any synchronization.
   */
  public void removeNoChecksumAnchor() {
    if (slot != null) {
      slot.removeAnchor();
    }
  }

  /**
   * Check if the replica has an associated mmap that has been fully loaded.
   *
   * Must be called with the cache lock held.
   */
  @VisibleForTesting
  public boolean hasMmap() {
    return ((mmapData != null) && (mmapData instanceof MappedByteBuffer));
  }

  /**
   * Free the mmap associated with this replica.
   *
   * Must be called with the cache lock held.
   */
  void munmap() {
    MappedByteBuffer mmap = (MappedByteBuffer)mmapData;
    // 调用 munmap() 系统调用移除映射
    NativeIO.POSIX.munmap(mmap);
    mmapData = null;
  }

  /**
   * 当一个ShortCircuitReplica对象的refCount等于0时，也就是没有任何对象引用这个ShortCircuitReplica对象时，就可以调用ShortCircuitReplica.close()方法关闭这个ShortCircuitReplica对象了。
   *
   * ShortCircuitReplica.close()方法首先会调用munmap系统调用删除数据块文件在内存中的映射数据，然后调用ShortCircuitCache.scheduleSlotReleaser()方法释放副本对应的槽位。
   */
  void close() {
    String suffix = "";
    
    Preconditions.checkState(refCount == 0, "tried to close replica with refCount " + refCount + ": " + this);
    refCount = -1;
    Preconditions.checkState(purged, "tried to close unpurged replica " + this);
    if (hasMmap()) {
      munmap(); // 调用munmap系统调用删除数据块文件在内存中的映射数据
      suffix += "  munmapped.";
    }
    IOUtils.cleanup(LOG, dataStream, metaStream);
    if (slot != null) {
      // 释放副本对应的槽位
      cache.scheduleSlotReleaser(slot); // TODO 核心代码
      suffix += "  scheduling " + slot + " for later release.";
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("closed " + this + suffix);
    }
  }

  public FileInputStream getDataStream() {
    return dataStream;
  }

  public FileInputStream getMetaStream() {
    return metaStream;
  }

  public BlockMetadataHeader getMetaHeader() {
    return metaHeader;
  }

  public ExtendedBlockId getKey() {
    return key;
  }

  public ClientMmap getOrCreateClientMmap(boolean anchor) {
    return cache.getOrCreateClientMmap(this, anchor); //
  }

  /**
   * loadMmapInternal()方法用于将ShortCircuitReplica对应的数据块文件映射到内存中，这个方法是在客户端对副本进行零拷贝读取时调用的，
   * 这个方法的逆方法是munmap()
   */
  MappedByteBuffer loadMmapInternal() {
    try {
      FileChannel channel = dataStream.getChannel();
      // 调用 FileChannel.map()方法 将数据块文件映射到内存中
      MappedByteBuffer mmap = channel.map(MapMode.READ_ONLY, 0,  Math.min(Integer.MAX_VALUE, channel.size()));
      return mmap;
    } catch (IOException e) {
      LOG.warn(this + ": mmap error", e);
      return null;
    } catch (RuntimeException e) {
      LOG.warn(this + ": mmap error", e);
      return null;
    }
  }

  /**
   * Get the evictable time in nanoseconds.
   *
   * Note: you must hold the cache lock to call this function.
   *
   * @return the evictable time in nanoseconds.
   */
  public Long getEvictableTimeNs() {
    return evictableTimeNs;
  }

  /**
   * Set the evictable time in nanoseconds.
   *
   * Note: you must hold the cache lock to call this function.
   *
   * @param evictableTimeNs   The evictable time in nanoseconds, or null
   *                          to set no evictable time.
   */
  void setEvictableTimeNs(Long evictableTimeNs) {
    this.evictableTimeNs = evictableTimeNs;
  }

  @VisibleForTesting
  public Slot getSlot() {
    return slot;
  }

  /**
   * Convert the replica to a string for debugging purposes.
   * Note that we can't take the lock here.
   */
  @Override
  public String toString() {
    return new StringBuilder().append("ShortCircuitReplica{").
        append("key=").append(key).
        append(", metaHeader.version=").append(metaHeader.getVersion()).
        append(", metaHeader.checksum=").append(metaHeader.getChecksum()).
        append(", ident=").append("0x").
          append(Integer.toHexString(System.identityHashCode(this))).
        append(", creationTimeMs=").append(creationTimeMs).
        append("}").toString();
  }
}
